package com.zhiyou.app.common.ws;

import lombok.Getter;
import lombok.Setter;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;

/**
 * 基于组合模式的可监听AtomicReference
 * 不继承AtomicReference，而是内部持有其实例，避免final方法无法重写的问题
 */
public class ObservableAtomicReference<T> {
    // 内部使用AtomicReference存储数据
    private final AtomicReference<T> atomicRef;
    private String name;
    /**
     * -- SETTER --
     * 设置变化监听器
     */
    // 变化监听器：接收旧值和新值
    @Setter
    private Consumer<Change<T>> changeListener;

    //线程池自定义threadFactory
    private final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(10, 200, 0L,
            TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(100), new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setName("ObservableAtomicReference-" + name);
            thread.setDaemon(true);
            return thread;
        }
    });

    // 构造方法
    public ObservableAtomicReference(String name) {
        this.atomicRef = new AtomicReference<>();
        this.name = name;
    }

    /**
     * 自定义set方法，内部调用AtomicReference的set，同时触发监听
     */
    public void set(T newValue) {
        T oldValue = atomicRef.get();
        atomicRef.set(newValue);
        //只有值真正变化时才通知
        if (!isEqual(oldValue, newValue)) {
            threadPoolExecutor.execute(() -> {
                notifyChange(oldValue, newValue);
            });
        }
    }

    /**
     * 自定义compareAndSet方法，内部调用AtomicReference的实现，成功后触发监听
     */
    public boolean compareAndSet(T expect, T update) {
        boolean success = atomicRef.compareAndSet(expect, update);
        if (success && !isEqual(expect, update)) {
            notifyChange(expect, update);
        }
        return success;
    }

    /**
     * 包装AtomicReference的其他方法
     */
    public T get() {
        return atomicRef.get();
    }

    public T getAndSet(T newValue) {
        T oldValue = atomicRef.getAndSet(newValue);
        if (!isEqual(oldValue, newValue)) {
            notifyChange(oldValue, newValue);
        }
        return oldValue;
    }

    /**
     * 通知监听器值已变化
     */
    private void notifyChange(T oldValue, T newValue) {
        if (changeListener != null) {
            // 这里可以考虑提交到线程池异步执行，避免阻塞调用线程
            changeListener.accept(new Change<>(oldValue, newValue));
        }
    }

    /**
     * 判断两个值是否相等（处理null的情况）
     */
    private boolean isEqual(T a, T b) {
        if (a == b) return true;
        return a != null && a.equals(b);
    }

    /**
     * 封装变化信息
     */
    @Getter
    public static class Change<T> {
        private final T oldValue;
        private final T newValue;

        public Change(T oldValue, T newValue) {
            this.oldValue = oldValue;
            this.newValue = newValue;
        }

    }
}


